package com.stan.core.spark.ad.topn;

import com.stan.core.mapper.AdTopNProvinceDailyCountMapper;
import com.stan.core.mapper.factory.MapperFactory;
import com.stan.core.vo.AdTopNProvinceDailyCount;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;
import org.apache.spark.sql.Row;
import java.io.Serializable;
import java.util.Arrays;

/***
 * 算法思路:
 * 1.输入的数据格式为: (yyyyMMdd_province_city_aid,clickCount)
 * 将这个格式拆解为 row(yyyyMMdd,province,city,aid,clickCount) 通过Spark SQL存入Hive中
 * 2.再使用Spark SQL对Hive调用SQL语句完成
 */
public class ComputeProvinceTop10Function implements VoidFunction<JavaPairRDD<String, Integer>>,Serializable{
    @Override
    public void call(JavaPairRDD<String, Integer> stringIntegerJavaPairRDD) throws Exception {

        JavaRDD<Row> rowsRDD = stringIntegerJavaPairRDD.map(new Function<Tuple2<String, Integer>, Row>() {
            @Override
            public Row call(Tuple2<String, Integer> dateProvinceCityAid2clickCount) throws Exception {
                String dateProvinceCityAid = dateProvinceCityAid2clickCount._1;
                int clickCount = dateProvinceCityAid2clickCount._2;
                String[] strSplited = dateProvinceCityAid.split("_");
                String date = strSplited[0];
                String province = strSplited[1];
                String city = strSplited[2];
                String aid = strSplited[3];
                return RowFactory.create(date,province,city,aid,clickCount);
            }
        });

        // 添加到Hive中
        StructType schema = DataTypes.createStructType(Arrays.asList(
                DataTypes.createStructField("date",DataTypes.StringType,true),
                DataTypes.createStructField("province",DataTypes.StringType,true),
                DataTypes.createStructField("city",DataTypes.StringType,true),
                DataTypes.createStructField("aid",DataTypes.StringType,true),
                DataTypes.createStructField("clickCount",DataTypes.IntegerType,true)
        ));

        HiveContext hiveContext = new HiveContext(stringIntegerJavaPairRDD.context());
        DataFrame dailyAdClickCountByProvinceDF = hiveContext.createDataFrame(rowsRDD,schema);
        rowsRDD.foreach(new VoidFunction<Row>() {
            @Override
            public void call(Row row) throws Exception {
                System.out.println("rowsRDD");
                System.out.println(row);
            }
        });
        // 注册成一张临时表
        dailyAdClickCountByProvinceDF.registerTempTable("tmp_date_city_aid_clickCount");

        // 对临时表进行查询，查询获得Top10的各个省份热门广告
        String sql = "select date,province,aid,clickCount,rank from (select date,province,aid,clickCount,ROW_NUMBER()" +
                " OVER ( PARTITION BY province ORDER BY clickCount DESC) as rank from tmp_date_city_aid_clickCount ) as t " +
                "where rank <= 10";
        DataFrame top10DF = hiveContext.sql(sql);
        JavaRDD<Row> resultRDD = top10DF.javaRDD();
        System.out.println("resultRDD:"+resultRDD.count());
        // 将Top10的数据存入MySQL
        resultRDD.foreach(new VoidFunction<Row>() {
            @Override
            public void call(Row row) throws Exception {
                // 提取
                String date = row.getString(0);
                String province = row.getString(1);
                String aid = row.getString(2);
                long clickCount = row.getInt(3);
                int rank = row.getInt(4);

                AdTopNProvinceDailyCount adTopNProvinceDailyCount = new AdTopNProvinceDailyCount();
                adTopNProvinceDailyCount.setDate(date);
                adTopNProvinceDailyCount.setProvince(province);
                adTopNProvinceDailyCount.setAid(aid);
                adTopNProvinceDailyCount.setClickCount(clickCount);

                AdTopNProvinceDailyCountMapper mapper = MapperFactory.getMapperFactory()
                        .getMapper(AdTopNProvinceDailyCountMapper.class);
                AdTopNProvinceDailyCount originData = mapper.getOne(adTopNProvinceDailyCount);
                // 如果 当天当省当rank的数据已经有了，就将它覆盖掉
                if(originData != null){
                    mapper.update(adTopNProvinceDailyCount);
                }else{ //如果没有，则进行插入
                    mapper.insert(adTopNProvinceDailyCount);
                }
            }
        });
    }
}


